package com.study;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.remoting.zookeeper.StateListener;
import org.apache.dubbo.remoting.zookeeper.curator.CuratorZookeeperClient;
import org.junit.Test;

import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.TimeUnit;

import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;

/**
 *
 * @Auther: luobiao
 * @Date: 2021/3/27 18:09
 * @Description:
 */
public class MyCuratorZookeeperClient {
    public final CuratorFramework client;
    //连接超时时间
    protected int DEFAULT_CONNECTION_TIMEOUT_MS = 3 * 1000;
    //会话超时时间，这个时间只是客户端定义的超时时间，最终要和服务器协商才能得出最终的会话超时时间（该协商值在连接成功后会被打印如："Negotiated session timeout: 40000"），协商逻辑如下
    //情况1: zoo.cfg 配置文件配置了maxSessionTimeOut和minSessionTimeOut，最终SessionTimeOut必须在minSessionTimeOut和maxSessionTimeOut区间里，如果跨越上下界，则以跨越的上届或下界为准。
    //情况2:配置文件没有配置maxSessionTimeOut和minSessionTimeOut，maxSessionTimeout没配置则 maxSessionTimeOut设置为 20 * tickTime，minSessionTimeOut没配置则 minSessionTimeOut设置为 2 * tickTime
    //也就是默认配置中tickTime为2秒时 SessionTimeOut的合法范围为 4秒~40秒。最终客户端和服务器端的心跳间隔为协商后的SessionTimeOut*(1/3)
    protected int DEFAULT_SESSION_TIMEOUT_MS = 60 * 1000;
    private static final String ZK_SESSION_EXPIRE_KEY = "zk.session.expire";
    private final Set<StateListener> stateListeners = new CopyOnWriteArraySet<StateListener>();

    public  MyCuratorZookeeperClient(URL url){
        try {
            int timeout = url.getParameter(TIMEOUT_KEY, DEFAULT_CONNECTION_TIMEOUT_MS);
            int sessionExpireMs = url.getParameter(ZK_SESSION_EXPIRE_KEY, DEFAULT_SESSION_TIMEOUT_MS);
            CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
                    .connectString(url.getBackupAddress())
                    //重试次数
                    .retryPolicy(new RetryNTimes(1, 1000))
                    .connectionTimeoutMs(timeout)
                    .sessionTimeoutMs(sessionExpireMs);
            String authority = url.getAuthority();
            if (authority != null && authority.length() > 0) {
                builder = builder.authorization("digest", authority.getBytes());
            }
            client = builder.build();
            client.getConnectionStateListenable().addListener(new MyCuratorZookeeperClient.CuratorConnectionStateListener(url));
            client.start();
            boolean connected = client.blockUntilConnected(timeout, TimeUnit.MILLISECONDS);
            if (!connected) {
                throw new IllegalStateException("zookeeper not connected");
            }
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }


    private class CuratorConnectionStateListener implements ConnectionStateListener {
        private final long UNKNOWN_SESSION_ID = -1L;

        private long lastSessionId;
        private URL url;

        public CuratorConnectionStateListener(URL url) {
            this.url = url;
        }

        @Override
        public void stateChanged(CuratorFramework client, ConnectionState state) {
            int timeout = url.getParameter(TIMEOUT_KEY, DEFAULT_CONNECTION_TIMEOUT_MS);
            int sessionExpireMs = url.getParameter(ZK_SESSION_EXPIRE_KEY, DEFAULT_SESSION_TIMEOUT_MS);

            long sessionId = UNKNOWN_SESSION_ID;
            try {
                sessionId = client.getZookeeperClient().getZooKeeper().getSessionId();
                System.out.println("sessionId=" + sessionId);
            } catch (Exception e) {
                System.out.println("获取session实例失败");
            }

            if (state == ConnectionState.LOST) {
                System.out.println("会话过期，上一个会话sessionId="+lastSessionId);
                MyCuratorZookeeperClient.this.stateChanged(StateListener.SESSION_LOST);
            } else if (state == ConnectionState.SUSPENDED) {
                System.out.println("会话" + Long.toHexString(sessionId) + " 连接超时. " +
                        "超时时间为： " + timeout + ", 会话过期时间为： " + sessionExpireMs);
                MyCuratorZookeeperClient.this.stateChanged(StateListener.SUSPENDED);
            } else if (state == ConnectionState.CONNECTED) {
                lastSessionId = sessionId;
                System.out.println("会话创建成功，sessionId=" + Long.toHexString(sessionId));
                MyCuratorZookeeperClient.this.stateChanged(StateListener.CONNECTED);
            } else if (state == ConnectionState.RECONNECTED) {
                //通过判断sessionId是不是发生变化了来确定重连后是不是创建了一个新的会话
                if (lastSessionId == sessionId && sessionId != UNKNOWN_SESSION_ID) {
                    System.out.println("zookeeper从会话过期中恢复，复用旧的sessionId =" + Long.toHexString(sessionId));
                    MyCuratorZookeeperClient.this.stateChanged(StateListener.RECONNECTED);
                } else {
                    System.out.println("旧会话丢失后创建新会话,旧的sessionId= " + Long.toHexString(lastSessionId) + "新的sessionId= " + Long.toHexString(sessionId));
                    lastSessionId = sessionId;
                    MyCuratorZookeeperClient.this.stateChanged(StateListener.NEW_SESSION_CREATED);
                }
            }
        }
    }


    protected void stateChanged(int state) {
        for (StateListener sessionListener : getSessionListeners()) {
            sessionListener.stateChanged(state);
        }
    }

    public Set<StateListener> getSessionListeners() {
        return stateListeners;
    }


    public void addStateListener(StateListener listener) {
        stateListeners.add(listener);
    }
}
